package com360.etc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class IndexStepTwo {
    public static class IndexStepTwoMapper extends Mapper<LongWritable, Text, Text, Text> {

        //产生<hello-文件名，1>
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {


            String[] split = value.toString().split("-");

            context.write(new Text(split[0]), new Text(split[1].replaceAll("\t", "-->")));


        }

    }

    public static class IndexStepTwoReducer extends Reducer<Text, Text, Text, Text> {

        //一组数据：<hello,a.txt-->4> <hello,b.txt-->4> <hello,c.txt-->4>
        @Override
        protected void reduce(Text key, Iterable<Text> values,
                              Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
            //StringBuffer是线程安全的，StringBuilder是非线程安全的，在不涉及线程安全下，StringBuilder更快
            StringBuilder sb = new StringBuilder();

            for (Text value : values) {
                sb.append(value.toString()).append("\t");

            }
            context.write(key, new Text(sb.toString()));


        }
    }

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        job.setJarByClass(IndexStepTwo.class);

        job.setMapperClass(IndexStepTwoMapper.class);
        job.setReducerClass(IndexStepTwoReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path("F:\\ddd"));
        FileOutputFormat.setOutputPath(job, new Path("F:\\fff"));


        job.waitForCompletion(true);






    }


}



